package com.batyuta.sort;

import org.apache.commons.io.FileUtils;
import org.apache.log4j.Logger;

import java.io.File;
import java.io.IOException;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Created by IntelliJ IDEA.
 * User: alexey
 * Date: 10/17/11
 * Time: 4:19 PM
 * To change this template use File | Settings | File Templates.
 */
public class MergeThreadPool {
    private static Logger logger = Logger.getLogger(MergeThreadPool.class);

    public static final int THREADS_COUNT = 10;
    /*
    * create a thread pool with four threads
    */
    ExecutorService execSvc = Executors.newFixedThreadPool(THREADS_COUNT * THREADS_COUNT);
    volatile ConcurrentLinkedQueue<MergeQueueItem> queueFirst = new ConcurrentLinkedQueue<MergeQueueItem>();
    volatile ConcurrentLinkedQueue<MergeQueueItem> queueDivision = new ConcurrentLinkedQueue<MergeQueueItem>();
    volatile ConcurrentLinkedQueue<MergeQueueItem> queueSort = new ConcurrentLinkedQueue<MergeQueueItem>();
    volatile ConcurrentLinkedQueue<MergeQueueItem> queueMerge1 = queueSort;
    volatile ConcurrentLinkedQueue<MergeQueueItem> queueMerge2 = new ConcurrentLinkedQueue<MergeQueueItem>();
    CountDownLatch countDownDivisionDone = new CountDownLatch(1);
    CountDownLatch countDownSortDone = new CountDownLatch(THREADS_COUNT);
    CountDownLatch countDownMergeDone;

    protected Integer lineCount;
    protected File inputFile;
    protected File outputFile;
    protected int dept = 1000;

    public MergeThreadPool(File inputFile, File outputFile, Integer lineCount) {
        this.inputFile = inputFile;
        this.lineCount = lineCount;
        this.outputFile = outputFile;
    }

    @Deprecated
    public MergeThreadPool() {
    }

    // TODO: delete temp files after each iteration
    public void run() {
        try {
            queueFirst.add(new MergeQueueItem(inputFile, null, lineCount, dept));
            MergeThreadDivision division = new MergeThreadDivision(queueFirst, queueDivision, null, countDownDivisionDone);
            execSvc.submit(division);
            for (int i = 0; i < THREADS_COUNT; i++) {
                MergeThreadSort sort = new MergeThreadSort(queueDivision, queueSort, countDownDivisionDone, countDownSortDone);
                execSvc.submit(sort);
            }

            countDownSortDone.await();

            do {
                int trdCount = queueMerge1.size() / 2;
                trdCount = THREADS_COUNT > trdCount ? trdCount : THREADS_COUNT;
                countDownMergeDone = new CountDownLatch(trdCount);
                for (int i = 0; i < trdCount; i++) {
                    MergeThreadMerge merge = new MergeThreadMerge(queueMerge1, queueMerge2, null, countDownMergeDone);
                    execSvc.submit(merge);
                }
                logger.info("waiting merging");
                countDownMergeDone.await();
                queueMerge1 = queueMerge2;
                queueMerge2 = new ConcurrentLinkedQueue<MergeQueueItem>();
            }
            while (queueMerge1.size() > 1);

            // last result
            MergeThreadMerge merge = new MergeThreadMerge(queueMerge1, queueMerge2, null, countDownMergeDone);
            execSvc.submit(merge);
            countDownMergeDone.await();
            MergeQueueItem item;
            while (null == (item = queueMerge2.poll())) {
                Thread.yield();
            }
            if (null != item) {
                File lastFile = item.getFileFirst();
                if (outputFile.exists()) {
                    FileUtils.forceDelete(outputFile);
                }
                String lastDirName = lastFile.getParentFile().getName();
                FileUtils.moveFile(lastFile, outputFile);
                for (int i = Integer.parseInt(lastDirName); i >= dept; i--) {
                    FileUtils.deleteDirectory(new File(inputFile.getParent() + "/" + i));
                }
            }

        } catch (IOException e) {
            logger.error(e, e);
        } catch (InterruptedException e) {
            logger.error(e, e);
        } finally {
            execSvc.shutdown();
            logger.info("End");
        }
    }

    public static void main(String... argv) throws InterruptedException {
        MergeThreadPool mergeThreadPool = new MergeThreadPool();
        mergeThreadPool.run();
    }
}
